
package com.windy.kafka.trace.producer;


import com.windy.common.transmit.kafka.KafkaProducerTransmitHeader;
import org.apache.kafka.clients.producer.Producer;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.kafka.core.ProducerFactory;

import java.util.List;

public class TracingProducerFactory<K, V> implements ProducerFactory<K, V>, DisposableBean {

    private final ProducerFactory<K, V> producerFactory;
    private List<MessageSendProcessor> tracers;
    private KafkaProducerTransmitHeader kafkaProducerTransmitHeader;

    public TracingProducerFactory(ProducerFactory<K, V> producerFactory, List<MessageSendProcessor> tracers, KafkaProducerTransmitHeader kafkaProducerTransmitHeader) {
        this.producerFactory = producerFactory;
        this.tracers = tracers;
        this.kafkaProducerTransmitHeader=kafkaProducerTransmitHeader;
    }


    @Override
    public Producer<K, V> createProducer() {
        return new TracingKafkaProducer<>(producerFactory.createProducer(), tracers,kafkaProducerTransmitHeader);
    }

    @Override
    public boolean transactionCapable() {
        return producerFactory.transactionCapable();
    }

    @Override
    public void destroy() throws Exception {
        if (producerFactory instanceof DisposableBean) {
            ((DisposableBean) producerFactory).destroy();
        }
    }
}
